//
// File:    bte_coll_base.C
// Author:  Octavian Procopiuc <tavi@cs.duke.edu>
//          (using some code by Rakesh Barve)
//
// Implementation of class BTE_collection_base.
//

#include <versions.H>
VERSION(bte_coll_base_C, "$Id: bte_coll_base.C,v 1.9 2002/07/20 21:36:55 tavi Exp $");

// Library configuration (logging, assertions, unistd.h, etc).
#include "lib_config.H"

// The BTE_collection_base class definition.
#include <bte_coll_base.H>

// For open(), read() and write().
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#if defined(__sun__)
extern int directio(int, int);
#endif

BTE_collection_header::BTE_collection_header(): 
  magic_number(BTE_COLLECTION_HEADER_MAGIC_NUMBER), 
  version(1), 
  type(0),
  header_length(sizeof(BTE_collection_header)), 
  total_blocks(1), 
  last_block(1), 
  used_blocks(0) {

#ifdef _SC_PAGE_SIZE
  os_block_size = sysconf(_SC_PAGE_SIZE);
#else    
  os_block_size = getpagesize(); 
#endif
  block_size = os_block_size;
}

tpie_stats_collection BTE_collection_base::gstats_;

void BTE_collection_base::create_stack() {
  // Fill in the stack file name.
  char stack_name[BTE_COLLECTION_PATH_NAME_LEN];
  strncpy((char *) stack_name, base_file_name_, BTE_COLLECTION_PATH_NAME_LEN - 4);
  strcat((char *) stack_name, BTE_COLLECTION_STK_SUFFIX);

  // Construct the pre-existing freeblock_stack.
  freeblock_stack_ = new stdio_stack<bid_t>((char *) stack_name, 
		       read_only_? BTE_READ_STREAM: BTE_WRITE_STREAM);
  
}

void BTE_collection_base::remove_stack_file() {
   // Fill in the stack file name.
  char stack_name[BTE_COLLECTION_PATH_NAME_LEN];
  strncpy((char *) stack_name, base_file_name_, BTE_COLLECTION_PATH_NAME_LEN - 4);
  strcat((char *) stack_name, BTE_COLLECTION_STK_SUFFIX);

  unlink(stack_name);
}

BTE_collection_base::BTE_collection_base(const char *base_name, 
		 BTE_collection_type type, size_t logical_block_factor):
  header_() {

  if (base_name == NULL) {
    status_ = BTE_COLLECTION_STATUS_INVALID;
    LOG_FATAL_ID("NULL file name passed to constructor");
    return;
  }
  
  strncpy((char*) base_file_name_, base_name, BTE_COLLECTION_PATH_NAME_LEN - 4);

  // A collection with a given name is not deleted upon destruction.
  per_ = PERSIST_PERSISTENT;

  shared_init(type, logical_block_factor);
}


void BTE_collection_base::shared_init(BTE_collection_type type,
				      size_t logical_block_factor) {
  read_only_ = (type == BTE_READ_COLLECTION);
  status_ = BTE_COLLECTION_STATUS_VALID;
  in_memory_blocks_ = 0;
  file_pointer = -1;

#ifdef _SC_PAGE_SIZE
  os_block_size_ = sysconf(_SC_PAGE_SIZE);
#else    
  os_block_size_ = getpagesize(); 
#endif

  // Fill in the blocks file name.
  char bcc_name[BTE_COLLECTION_PATH_NAME_LEN];
  strncpy((char *) bcc_name, base_file_name_, BTE_COLLECTION_PATH_NAME_LEN - 4);  
  strcat((char *) bcc_name, BTE_COLLECTION_BLK_SUFFIX);

  if (read_only_) {

    if ((bcc_fd_ = ::open(bcc_name, O_RDONLY)) == -1) {
      status_ = BTE_COLLECTION_STATUS_INVALID;
      LOG_FATAL_ID("open() failed to open read-only file: ");
      LOG_FATAL_ID(bcc_name);	
      return;
    }

    if (read_header(bcc_name) != BTE_ERROR_NO_ERROR) {
      status_ = BTE_COLLECTION_STATUS_INVALID;
      return;
    }

    // Check whether we need a stack.
    if (header_.used_blocks < header_.last_block - 1) {
      create_stack();
      if (freeblock_stack_->status() == BTE_STREAM_STATUS_INVALID) {
	status_ = BTE_COLLECTION_STATUS_INVALID;
	return;
      }
    } else
      freeblock_stack_ = NULL;

  } else  {   // Writeable bcc.

    // If a new collection, remove any existing files with the same names.
    if (type == BTE_WRITE_NEW_COLLECTION) {
      unlink(bcc_name);
      remove_stack_file();
    }
      
    // Open the file for writing.  First we will try to open 
    // it with the O_EXCL flag set.  This will fail if the file
    // already exists.  If this is the case, we will call open()
    // again without it and read in the header block.
    if ((bcc_fd_ = ::open(bcc_name, O_RDWR | O_CREAT | O_EXCL,
             S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH)) == -1) {
      
      // Try again, hoping the file already exists.
      if ((bcc_fd_ = ::open(bcc_name, O_RDWR)) == -1) {
        status_ = BTE_COLLECTION_STATUS_INVALID;        
        LOG_FATAL_ID("open() failed to open file:");
	LOG_FATAL_ID(bcc_name);
        return;
      }
      
      if (read_header(bcc_name) != BTE_ERROR_NO_ERROR) {
	status_ = BTE_COLLECTION_STATUS_INVALID;
	return;
      }

      // Check whether we need a stack.
      if (header_.used_blocks < header_.last_block - 1) {
	create_stack();
	if (freeblock_stack_->status() == BTE_STREAM_STATUS_INVALID) {
	  status_ = BTE_COLLECTION_STATUS_INVALID;
	  return;
	}
      } else 
	freeblock_stack_ = NULL;

    } else {   // The file was just created.
      
      tp_assert(header_.magic_number == BTE_COLLECTION_HEADER_MAGIC_NUMBER, "Header magic number mismatch.");
      tp_assert(header_.os_block_size == os_block_size_, "Header os_block_size mismatch.");

      header_.block_size = logical_block_factor * header_.os_block_size;

      if (write_header(bcc_name) != BTE_ERROR_NO_ERROR) {
	status_ = BTE_COLLECTION_STATUS_INVALID;
	return;
      }

      // No stack (yet). Will be created by delete if needed.
      freeblock_stack_ = NULL;

      gstats_.record(COLLECTION_CREATE);
      stats_.record(COLLECTION_CREATE);
    }
  }

#if defined(__sun__) 
  if (direct_io)
    directio(bcc_fd_, DIRECTIO_ON);
  else
    directio(bcc_fd_, DIRECTIO_OFF);
#endif

  gstats_.record(COLLECTION_OPEN);
  stats_.record(COLLECTION_OPEN);
}

#if defined(__sun__) 
bool BTE_collection_base::direct_io = false;
#endif


BTE_err BTE_collection_base::read_header(char* bcc_name) {

  char * tmp_buffer = new char[os_block_size_];

  if (::lseek(bcc_fd_, 0, SEEK_SET) != 0) {
    LOG_FATAL_ID("Failed to lseek in file:");
    LOG_FATAL_ID(bcc_name);
    return BTE_ERROR_IO_ERROR;
  }

  if (::read(bcc_fd_, (char *)tmp_buffer, os_block_size_) != os_block_size_) {
    LOG_FATAL_ID("Failed to read() in file:");
    LOG_FATAL_ID(bcc_name);
    return BTE_ERROR_IO_ERROR;
  }

  file_pointer = os_block_size_;

  memcpy((void *) &header_, (const void *) tmp_buffer, 
	 sizeof(BTE_collection_header));
  delete [] tmp_buffer;

  // Do some error checking on the header, such as to make sure that
  // it has the correct header version, block size etc.
  if (header_.magic_number != BTE_COLLECTION_HEADER_MAGIC_NUMBER || 
      header_.os_block_size != os_block_size_) {
    LOG_FATAL_ID("Invalid header in file: ");
    LOG_FATAL_ID(bcc_name);
    return BTE_ERROR_BAD_HEADER;
  }
    
  off_t lseek_retval;
  // Some more error checking.
  if ((lseek_retval = ::lseek(bcc_fd_, 0, SEEK_END)) != bid_to_file_offset(header_.total_blocks)) {
    LOG_FATAL_ID("Header field total_blocks incompatible with Unix file length in file:");
    LOG_FATAL_ID(bcc_name);
    return BTE_ERROR_BAD_HEADER;
  }

  file_pointer = lseek_retval;

  return BTE_ERROR_NO_ERROR;
}


BTE_err BTE_collection_base::write_header(char *bcc_name) {

  char * tmp_buffer = new char[os_block_size_];
  memcpy((void *) tmp_buffer, (const void *) &header_, 
	 sizeof(BTE_collection_header));

  if (::lseek(bcc_fd_, 0, SEEK_SET) != 0) {
    LOG_FATAL_ID("Failed to lseek() in file:");
    LOG_FATAL_ID(bcc_name);
    return BTE_ERROR_IO_ERROR;
  }

  if (::write(bcc_fd_, tmp_buffer, os_block_size_) != os_block_size_) {
    LOG_FATAL_ID("Failed to write() in file:");
    LOG_FATAL_ID(bcc_name);
    return BTE_ERROR_IO_ERROR;
  }

  file_pointer = os_block_size_;

  delete [] tmp_buffer;
  return BTE_ERROR_NO_ERROR;
}


BTE_collection_base::~BTE_collection_base() {

  char bcc_name[BTE_COLLECTION_PATH_NAME_LEN];
  strncpy((char *) bcc_name, base_file_name_, 
	  BTE_COLLECTION_PATH_NAME_LEN - 4);
  strcat((char *) bcc_name, BTE_COLLECTION_BLK_SUFFIX);

  // No block should be in memory at the time of destruction.
  if (in_memory_blocks_) {
    LOG_WARNING_ID("In memory blocks when closing collection in:");
    LOG_WARNING_ID(base_file_name_);
  }

#if defined(__sun__) 
  if (direct_io)
    directio(bcc_fd_, DIRECTIO_OFF);
#endif

  // Write the header.
  if (!read_only_)
    write_header(bcc_name);

  // Delete the stack.
  if (freeblock_stack_ != NULL) {
    freeblock_stack_->persist(per_);
    delete freeblock_stack_;
  }

  // Close the blocks file.
  if (::close(bcc_fd_)) {            
    LOG_FATAL_ID("Failed to close() ");
    LOG_FATAL_ID(bcc_name);
    return;
  }
  
  // If necessary, remove the blocks file.
  if (per_ == PERSIST_DELETE) {
    if (read_only_) {
      LOG_WARNING_ID("Read-only collection is PERSIST_DELETE");
      LOG_WARNING_ID(bcc_name);
      return;
    } 

    if (unlink(bcc_name)) {
      LOG_FATAL_ID("Failed to unlink() ");
      LOG_FATAL_ID(bcc_name);
      return;
    } else {
      gstats_.record(COLLECTION_DELETE);
      stats_.record(COLLECTION_DELETE);
    }
  }

  gstats_.record(COLLECTION_CLOSE);
  stats_.record(COLLECTION_CLOSE);
}
